package com.chb.flink.func

import java.text.SimpleDateFormat
import java.util.Date

import com.chb.flink.source.StationLog
import org.apache.flink.api.common.functions.MapFunction
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment

/**
 * 普通函数类举例：按照指定的时间格式输出每个通话的拨号时间和结束时间
 */
object TestFunction {
    def main(args: Array[String]): Unit = {
        val streamEnv = StreamExecutionEnvironment.getExecutionEnvironment
        streamEnv.setParallelism(1)

        // 隐式转换
        import org.apache.flink.streaming.api.scala._

        val data: DataStream[StationLog] = streamEnv.readTextFile(getClass.getResource("/station.log").getPath)
            .map(line => {
                val arr = line.split(",")
                new StationLog(arr(0).trim, arr(1).trim, arr(2).trim, arr(3).trim, arr(4).trim.toLong, arr(5).trim.toLong)
            })

        //定义时间输出格式
        val format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")
        //过滤那些通话成功的
        data.filter(_.callType.equals("success"))
            .map(new CallMapFunction(format))
            .print()
        streamEnv.execute()
    }
}

//自定义的函数类
class CallMapFunction(format: SimpleDateFormat) extends MapFunction[StationLog, String] {
    override def map(t: StationLog): String = {
        var strartTime = t.callTime;
        var endTime = t.callTime + t.duration * 1000
        "主叫号码:" + t.callOut + ",被叫号码:" + t.callIn + ",呼叫起始时间:" + format.format(new Date(strartTime)) + ",呼叫结束时间:" + format.format(new Date(endTime))
    }
}

